import javafx.scene.control.Tab;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;

public class Demo3Dianxin {
    Configuration conf = null;
    Connection conn = null;

    @Before
    public void init() {
        conf = HBaseConfiguration.create();
        // 其他配置参考http://hbase.apache.org/1.4/book.html#config.files
        conf.set("hbase.zookeeper.quorum", "master:2181,node1:2181,node2:2181");
        try {
            conn = ConnectionFactory.createConnection(conf);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Test
    public void loadDataToHBase() throws IOException {
        Admin admin = conn.getAdmin();
        // 判断表是否存在，不存在即创建
        if (!admin.tableExists(TableName.valueOf("dianxin"))) {
            HTableDescriptor dianxin = new HTableDescriptor(TableName.valueOf("dianxin"));
            HColumnDescriptor info = new HColumnDescriptor("info");
            info.setMaxVersions(10); // 设置保存10个版本
            dianxin.addFamily(info);
            admin.createTable(dianxin);
        }

        Table dianxin = conn.getTable(TableName.valueOf("dianxin"));
        // 读取文件
        BufferedReader br = new BufferedReader(new FileReader("data/DIANXIN.csv"));
        String line = br.readLine();
        while (line != null) {
            String[] splits = line.split(",");
            String rk = splits[0];
            // 进入网格的时间
            String start_time = splits[1];
            // 经度
            String lg = splits[4];
            // 维度
            String lat = splits[5];

            Put put = new Put(Bytes.toBytes(rk));
            // 以start_time日期字符串作为"时间戳"
            // 并不需要转成"时间戳"，只要能保持最新的数据的"时间戳"更大就可以了
            put.addColumn("info".getBytes(), "lg".getBytes(), Long.parseLong(start_time), lg.getBytes());
            put.addColumn("info".getBytes(), "lat".getBytes(), Long.parseLong(start_time), lat.getBytes());

            dianxin.put(put);

            line = br.readLine();
        }
        br.close();
    }

    @Test
    public void getLast10Pos() throws IOException {
        Table dianxin = conn.getTable(TableName.valueOf("dianxin"));
        String mdn = "C7BDBE02C41C8E9F69668DE3A0867E3776FB9A6F";

        Get get = new Get(mdn.getBytes());
        // 如果需要获取多个版本，需要手动设置，默认获取最新的
        get.setMaxVersions(10);

        Result rs = dianxin.get(get);
        for (Cell cell : rs.listCells()) {
            String rk = Bytes.toString(CellUtil.cloneRow(cell));
            String cf = Bytes.toString(CellUtil.cloneFamily(cell));
            String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
            String value = Bytes.toString(CellUtil.cloneValue(cell));

            System.out.println(rk + "\t" + cf + "\t" + qualifier + "\t" + value);
        }

    }

    @After
    public void closeAll() throws IOException {
        if (conn != null) {
            conn.close();
        }
    }
}
